package http.executor;

import com.alibaba.fastjson.JSON;
import http.common.ResponseEntity;
import http.observer.Observer;
import http.observer.impl.ContextMapObserver;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class ResponseExecutor {

    private static Map<String, ChannelHandlerContext> contextMap=new ConcurrentHashMap<>();

    private static List<Observer> observers=new ArrayList<>();

    static {
        observers.add(new ContextMapObserver());
        ExecutorService pool = Executors.newCachedThreadPool();
        pool.execute(new Runnable() {
            @Override
            public void run() {
                exec();
            }
        });

    }

    private static LinkedBlockingQueue<ResponseEntity> messageQueue = new LinkedBlockingQueue<>();



    public static void putResult(String id,Object result){
        try {
            messageQueue.put(new ResponseEntity(id,result));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void exec() {
        while (true){
            try {
                ResponseEntity entity=messageQueue.take();
                ChannelHandlerContext ctx = contextMap.remove(entity.getId());
                if(ctx == null){
                    continue;
                }
                FullHttpResponse response = responseOK(HttpResponseStatus.OK, JSON.toJSONString(entity.getResult()));
                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
            }catch (Exception e){
                e.printStackTrace();
            }

        }
    }

    public static void putChannelHandlerContext(String id,ChannelHandlerContext ctx){
        if(id.length() < 20){
            System.out.println("id : "+id +"非法");
        }
        contextMap.put(id,ctx);
        notify(id);//通知观察者
    }

    private static void notify(String id){
        for (Observer observer : observers) {
            observer.update(id);
        }
    }

    private static FullHttpResponse responseOK(HttpResponseStatus status, String content) {
        ByteBuf byteBuf = Unpooled.copiedBuffer(content.getBytes());
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, byteBuf);
        if (content != null) {
            response.headers().set("Content-Type", "text/plain;charset=UTF-8");
            response.headers().set("Content_Length", response.content().readableBytes());
        }
        return response;
    }

    public static Map<String, ChannelHandlerContext> getContextMap() {
        return contextMap;
    }

    public static void setContextMap(Map<String, ChannelHandlerContext> contextMap) {
        ResponseExecutor.contextMap = contextMap;
    }
}
